package api.sink;

import api.beans.SensorReading;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.util.Collector;

/**
 * Sink 输出文件
 */
public class SinkTest3_file {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String formPath = "D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt";
        String toPathBase = "D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\outData";

        DataStream<String> inputStream = env.readTextFile(formPath);
        // 转POJO类
        DataStream<SensorReading> dataStream1 = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });
        // 转二元组
        DataStream dataStream2 = inputStream.flatMap(new FlatMapFunction<String, Tuple2<String, Double>>(){
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Double>> collector) throws Exception {
                String[] fields = s.split(",");
                collector.collect(new Tuple2<String, Double>(fields[0], Double.valueOf(fields[2])));
            }
        });

         dataStream2.addSink(
            StreamingFileSink.forRowFormat(
                new Path(toPathBase),
                new SimpleStringEncoder<SensorReading>()
            ).build()
        );

        env.execute();
    }
}
